package com.eb.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.eb.annotation.AutoAnnouncement;
import com.eb.annotation.AutoCalculateExp;
import com.eb.context.BaseContext;
import com.eb.domain.Look;
import com.eb.enumeration.OperationType;
import com.eb.mapper.ILookMapper;
import com.eb.service.LookService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class LookServiceImpl extends ServiceImpl<ILookMapper, Look> implements LookService {

    @Autowired
    StringRedisTemplate redisTemplate;
    //异步执行浏览记录的写入，创建线程池
    private static final ExecutorService LOOKS_EXECUTOR = Executors.newSingleThreadExecutor();

    @PostConstruct             //该注解为Spring提供，表示初始化容器之后就直接执行
    private void init()
    {
        //初始化之后立刻提交任务
        LOOKS_EXECUTOR.submit(new Looks_Handler());
    }
    //编写内部类来进行任务线程的执行
    private class Looks_Handler implements Runnable{

        @Override
        public void run() {
            while (true)
            {
                try {
                    //1.获取消息队列中的浏览信息
                    List<MapRecord<String, Object, Object>> list = redisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),    //指定消费者组+消费者
                            StreamReadOptions.empty().count(20).block(Duration.ofSeconds(5)),//读取选项 一次读取20条，5s一次
                            StreamOffset.create("stream.looks", ReadOffset.lastConsumed())//队列名，读取指针
                    );

                    //2.判断消息获取是否成功
                    if (list == null || list.isEmpty())
                    {
                        //2-1不存在，说明没有消息，继续下一次循环
                        log.info("未读取到消息");
                        continue;
                    }
                    log.info("读取到消息");
                    //解析消息，并封装
                    List<Look> looks = new ArrayList<>();
                    for (MapRecord<String, Object, Object> record:list)
                    {
                        //获取map集合
                        Map<Object, Object> map = record.getValue();
                        String userId = (String)map.get("user_id");
                        String innovationId = (String)map.get("innovation_id");
                        log.info("用户id：{} 帖子id:{}",userId,innovationId);
                        looks.add(new Look(null,Integer.valueOf(userId),Integer.valueOf(innovationId)
                        , LocalDateTime.now()));
                        //ACK确认，用完之后删除
                        log.info("消息已确认");
                        redisTemplate.opsForStream().acknowledge("stream.looks","g1",record.getId());
                        redisTemplate.opsForStream().delete("stream.looks",record.getId());
                        log.info("删除消息");
                    }
                    //2-2将解析到的消息批量插入数据库
                    saveBatch(looks);

                }catch (Exception e)
                {
                    log.error("获取异常{}",e);
                    //异常就去PendingList中进行获取，并再次进行处理
//                        handlePendingList();
                }
            }
        }

        private void handlePendingList()  {
            while (true)
            {
                try {
                    //1.获取PendingList中的浏览信息
                    List<MapRecord<String, Object, Object>> list = redisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),    //指定消费者组+消费者
                            StreamReadOptions.empty().count(20),
                            StreamOffset.create("stream.looks", ReadOffset.from("0"))//队列名，读取指针
                    );
                    //2.判断消息获取是否成功
                    if (list == null || list.isEmpty())
                    {
                        //2-1不存在，说明PendingList没有异常消息，继续下一次循环
                        log.info("未发现异常，跳出循环");
                        break;
                    }
                    log.info("读取到异常");
                    //解析消息，并封装
                    List<Look> looks = new ArrayList<>();
                    for (MapRecord<String, Object, Object> record:list)
                    {
                        //获取map集合
                        Map<Object, Object> map = record.getValue();
                        String userId = (String)map.get("user_id");
                        String innovationId = (String)map.get("innovation_id");
                        log.info("用户id：{} 帖子id:{}",userId,innovationId);
                        looks.add(new Look(null,Integer.valueOf(userId),Integer.valueOf(innovationId)
                                , LocalDateTime.now()));
                        //ACK确认，用完之后删除
                        redisTemplate.opsForStream().acknowledge("stream.looks","g1",record.getId());
                    }
                    //2-2将解析到的消息批量插入数据库
                    saveBatch(looks);

                }catch (Exception e)
                {
                    log.error("获取异常{}",e);
                    try {
                        Thread.sleep(50);
                    }catch (Exception e2)
                    {
                        e2.printStackTrace();
                    }

                }
            }
        }
    }



    @Override
    @AutoAnnouncement(OperationType.LOOK)
    @AutoCalculateExp(OperationType.LOOK)
    public boolean addLook(Integer innovationId) {
        Integer userId = BaseContext.getCurrentId();
        //先判断是否已经浏览过
        QueryWrapper<Look> wrapper = new QueryWrapper<>();
        wrapper.eq("user_id",userId)
                .eq("innovation_id",innovationId);
        Look look = getOne(wrapper);
        if (look != null)
        {
            return false;
        }
        //编辑Map便于传入消息队列
        Map<String,String> map = new HashMap<>();
        map.put("user_id", userId.toString());
        map.put("innovation_id",innovationId.toString());
        //向Redis中的Stream里发送消息
        redisTemplate.opsForStream().add("stream.looks",map);
        return true;
    }
}
